package org.apache.flink.runtime.resourceconsumer.factories;

import javax.annotation.Nonnull;

import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.runtime.metrics.groups.ConsumerManagerConsumerMetricGroup;
import org.apache.flink.runtime.metrics.groups.ConsumerManagerMetricGroup;

/**
 * Default implementation of {@link ConsumerManagerConsumerMetricGroupFactory} which creates for a given
 * {@link JobGraph} a {@link ConsumerManagerMetricGroup}.
 */
public class DefaultResourceConsumerResourceMetricGroupFactory implements ConsumerManagerConsumerMetricGroupFactory {

    private final ConsumerManagerMetricGroup consumerManagerMetricGroup;

    public DefaultResourceConsumerResourceMetricGroupFactory(
            @Nonnull ConsumerManagerMetricGroup consumerManagerMetricGroup) {
        this.consumerManagerMetricGroup = consumerManagerMetricGroup;
    }

    @Override
    public ConsumerManagerConsumerMetricGroup create(@Nonnull ConsumerID resourceId, String consumerName) {
        return consumerManagerMetricGroup.addConsumer(resourceId, consumerName);
    }
}
